package com.atguigu.flink.chapter10.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/*
30
    第一名   30

40
    第一名   40
    第二名   30

35
   第一名   40
   第二名   35
 */
public class AggTableFunctionDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql("create table sensor(" +
                "id string," +
                "ts bigint," +
                "vc int" +
                ") with(" +
                "  'connector' = 'kafka', " +
                "  'topic' = 's1', " +
                "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
                "  'properties.group.id' = 'atguigu', " +
                "  'scan.startup.mode' = 'latest-offset', " +
                "  'format' = 'csv' " +
                ")");

        Table table = tEnv.from("sensor");

        //1、注册一个自定义函数
        tEnv.createTemporaryFunction("top2",Top2.class);

        //2、在sql中使用
        table
                .groupBy($("id"))
                .flatAggregate(Expressions.call("top2",$("vc")))
                .select($("id"),$("rank"),$("vc"))
                .execute()
                .print();
    }

    public static class Top2 extends TableAggregateFunction<Result,FirstSecond> {

        @Override
        public FirstSecond createAccumulator() {
            return new FirstSecond();
        }

        //进行累加
        public void accumulate(FirstSecond fs,Integer vc){
            if (vc > fs.first){
                fs.second = fs.first;
                fs.first = vc;
            } else if (vc > fs.second) {
                fs.second = vc;
            }
        }

        //把结果输出
        //方法名：emitValue
        //参数1：累计器
        //参数2：收集器
        public void emitValue(FirstSecond fs, Collector<Result> out){
            out.collect(new Result("第一名", fs.first));
            if (fs.second > Integer.MIN_VALUE){
                out.collect(new Result("第二名", fs.second));
            }
        }
    }



    public static class Result {
        public String rank;
        public Integer vc;

        public Result(String rank , Integer vc){
            this.rank = rank;
            this.vc = vc;
        }
    }

    public static class FirstSecond {
        public Integer first = Integer.MIN_VALUE;
        public Integer second = Integer.MIN_VALUE;
    }
}


//public class AggTableFunctionDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int" +
//                ") with(" +
//                "  'connector' = 'kafka', " +
//                "  'topic' = 's1', " +
//                "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
//                "  'properties.group.id' = 'atguigu', " +
//                "  'scan.startup.mode' = 'latest-offset', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        Table table = tEnv.from("sensor");
//
//        //1、自定义一个函数
//        tEnv.createTemporaryFunction("top2",Top2.class);
//
//        //2、在sql中使用
//        table.groupBy($("id"))
//                .flatAggregate(Expressions.call("top2",$("vc")))
//                .select($("id"),$("rank"),$("vc"))
//                .execute()
//                .print();
//    }
//
//    private static class Top2 extends TableAggregateFunction<Result,FirstSecond>{
//        @Override
//        public FirstSecond createAccumulator() {
//            return new FirstSecond();
//        }
//
//        //进行累加
//        public void accumulate(FirstSecond fs,Integer vc){
//            if (vc > fs.first){
//                fs.second = fs.first;
//                fs.first = vc;
//            } else if (vc > fs.second) {
//                fs.second = vc;
//            }
//        }
//
//
//        //把结果输出
//        //方法名：emitValue
//        //参数1：累计器
//        //参数2：收集器
//        public void emitValue(FirstSecond fs,Collector<Result> out){
//            out.collect(new Result("第一名", fs.first));
//            if (fs.second > Integer.MIN_VALUE){
//                out.collect(new Result("第二名",fs.second));
//            }
//        }
//    }
//
//    public static class Result {
//        public String rank;
//        public Integer vc;
//        public Result(String rank,Integer vc){
//            this.rank = rank;
//            this.vc = vc;
//        }
//    }
//
//    public static class FirstSecond {
//        public Integer first = Integer.MIN_VALUE;
//        public Integer second = Integer.MIN_VALUE;
//    }
//
//}



//public class AggTableFunctionDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int" +
//                ") with(" +
//                "  'connector' = 'kafka', " +
//                "  'topic' = 's1', " +
//                "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
//                "  'properties.group.id' = 'atguigu', " +
//                "  'scan.startup.mode' = 'latest-offset', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        Table table = tEnv.from("sensor");
//
//        //1、注册一个自定义函数
//        tEnv.createTemporaryFunction("top2",Top2.class);
//
//        //2.在sql中使用
//        table.groupBy($("id"))
//                .flatAggregate(Expressions.call("top2",$("vc")))
//                .select($("id"),$("rank"),$("vc"))
//                .execute()
//                .print();
//    }
//
//
//    public static class Top2 extends TableAggregateFunction<Result,FirstSecond>{
//
//        @Override
//        public FirstSecond createAccumulator() {
//            return new FirstSecond();
//        }
//
//
//        //进行累加
//        public void accumulate(FirstSecond fs,Integer vc){
//            if (vc > fs.first){
//                fs.second = fs.first;
//                fs.first = vc;
//            } else if (vc > fs.second) {
//                fs.second = vc;
//            }
//        }
//
//
//        //把结果输出
//        //方法名：emitValue
//        //参数1：累计器
//        //参数2：收集器
//        public void emitValue(FirstSecond fs,Collector<Result> out){
//            out.collect(new Result("第一名",fs.first));
//            if (fs.second > Integer.MIN_VALUE){
//                out.collect(new Result("第二名",fs.second));
//            }
//        }
//    }
//
//
//
//    public static class Result {
//        public String rank;
//        public Integer vc;
//
//        public Result(String rank,Integer vc) {
//            this.rank = rank;
//            this.vc = vc;
//        }
//    }
//
//    public static class FirstSecond {
//        public Integer first = Integer.MIN_VALUE;
//        public Integer second = Integer.MIN_VALUE;
//    }
//}



//public class AggTableFunctionDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int" +
//                ") with(" +
//                "  'connector' = 'kafka', " +
//                "  'topic' = 's1', " +
//                "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
//                "  'properties.group.id' = 'atguigu', " +
//                "  'scan.startup.mode' = 'latest-offset', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        Table table = tEnv.from("sensor");
//
//        //1、注册一个自定义函数
//        tEnv.createTemporaryFunction("top2",Top2.class);
//
//        //2、在sql中使用
//        table.groupBy($("id"))
//                .flatAggregate(Expressions.call("top2",$("vc")))
//                .select($("id"),$("ts"),$("vc"))
//                .execute()
//                .print();
//
//    }
//
//    private static class Top2 extends TableAggregateFunction<Result,FirstSecond>{
//        @Override
//        public FirstSecond createAccumulator() {
//            return new FirstSecond();
//        }
//
//        //进行累加
//        public void acccumulate(FirstSecond fs,Integer vc){
//            if (vc > fs.first){
//                fs.second = fs.first;
//                fs.first = vc;
//            } else if (vc > fs.second) {
//                fs.second = vc;
//            }
//        }
//
//        //把结果输出
//        //方法名：emitValue
//        //参数1：累计器
//        //参数2：收集器
//        public void emitValue(FirstSecond fs,Collector<Result> out){
//            out.collect(new Result("第一名",fs.first));
//            if (fs.second > Integer.MIN_VALUE){
//                out.collect(new Result("第二名",fs.second));
//            }
//        }
//    }
//
//    private static class FirstSecond {
//        public Integer first = Integer.MIN_VALUE;
//        public Integer second = Integer.MIN_VALUE;
//    }
//
//    private static class Result {
//        public String rank;
//        public Integer vc;
//
//        public Result(String rank, Integer vc) {
//            this.rank = rank;
//            this.vc = vc;
//        }
//    }
//}



//public class AggTableFunctionDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int" +
//                ") with(" +
//                "  'connector' = 'kafka', " +
//                "  'topic' = 's1', " +
//                "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
//                "  'properties.group.id' = 'atguigu', " +
//                "  'scan.startup.mode' = 'latest-offset', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        Table table = tEnv.from("sensor");
//
//        //1、注册一个自定义函数
//        tEnv.createTemporaryFunction("top2",Top2.class);
//
//        //2、在sql中使用
//        table.groupBy($("id"))
//                .flatAggregate(Expressions.call("top2",$("vc")))
//                .select($("id"),$("rank"),$("vc"))
//                .execute()
//                .print();
//    }
//
//    public static class Top2 extends TableAggregateFunction<Result,FirstSecond>{
//
//        @Override
//        public FirstSecond createAccumulator() {
//            return new FirstSecond();
//        }
//
//        //进行累加
//        public void acccumulate(FirstSecond fs , Integer vc){
//            if (vc > fs.first ){
//                fs.second = fs.first;
//                fs.first = vc;
//            } else if (vc > fs.second) {
//                fs.second = vc;
//            }
//        }
//
//        public void emitValue(FirstSecond fs,Collector<Result> out){
//            out.collect(new Result("第一名", fs.first));
//            if (fs.second > Integer.MIN_VALUE){
//                out.collect(new Result("第二名",fs.second));
//            }
//        }
//    }
//
//    public static class FirstSecond {
//        public Integer first = Integer.MIN_VALUE;
//        public Integer second = Integer.MIN_VALUE;
//    }
//
//    public static class Result {
//        public String rank;
//        public Integer vc;
//
//        public Result(String rank, Integer vc) {
//            this.rank = rank;
//            this.vc = vc;
//        }
//    }
//}